//
// Created by 郝羽 on 2018/6/25.
//

#include "conn.hpp"

#include <chrono>
#include <string>

constexpr char kContentTypeJson[] = "application/json";

constexpr size_t kPayloadTypeVp8 = 96;

// fix this: todo: use different ssrc
// this has to be same with rtc endpoint or we can't parse rtcp
constexpr uint32_t kVideoSSRC = 1843363638;
constexpr uint32_t kAudioSSRC = 754039198;

constexpr size_t kVideoFramesPerSecond = 10;
constexpr auto kVideoFramePeriod = std::chrono::milliseconds(1000 / kVideoFramesPerSecond);
constexpr auto kVideoKeyframePeriod = std::chrono::seconds(120);
constexpr uint32_t kVideoSampleRate = 90000;  // Hz
//constexpr auto kStatPeriod = std::chrono::seconds(1);
//constexpr auto kConnStatPeriod = std::chrono::milliseconds(200);
constexpr auto kAudioFramePeriod = std::chrono::milliseconds(20);
constexpr uint32_t kAudioSampleRate = 48000;

constexpr uint8_t kPayloadTypeOpus = 111;

bool Conn::on_stream_data(rtc_endpoint_t rtc, void *context, int stream_type,
        rtc_codec_id_t codec, rtc_codec_id_t fec_codec, int data_type,
        int is_rtcp, int is_rtx, unsigned char *buf, int len) {
    if (data_type != RTC_RECV_CLEAR) {
        return false;
    }

    auto connection = static_cast<Conn*>(context);

    if (is_rtcp) {
        connection->handle_incoming_rtcp(buf, len);
    } else {
        connection->handle_incoming_rtp(buf, len);
    }

    return false;
}

void Conn::on_stream_event(rtc_endpoint_t rtc, void *context, int event,
        void *event_arg1, int event_arg2) {
    auto connection = static_cast<Conn*>(context);
    if (event == RTC_EVENT_SETUP) {
        connection->handle_setup_event();
    }
}

Conn::Conn(Context &ctx, int rtc_id, ConnectionEventHandler handler, void* data)
        : ctx_(ctx), handler_(handler), handler_data_(data), rtc_id_(rtc_id) {
    rtcp_parser_ = rtcp_parser_create();
    _video_frame_size = std::chrono::duration_cast<std::chrono::milliseconds>(kVideoFramePeriod).count() * ctx.video_bitrate / 8000;
    if (!ctx.audios.empty()) {
        audio_ = ctx.audios.front();
        ctx.audios.pop_front();
        g_message("attached audio file %s", audio_->FilePathName().c_str());
    }
}

Conn::~Conn() {
    if (rtcp_parser_) {
        rtcp_parser_delete(rtcp_parser_);
        rtcp_parser_ = nullptr;
    }
    if (_callback_handle && _endpoint) {
        rtc_endpoint_unregister_callback(_endpoint, _callback_handle);
    }
    if (_endpoint) {
        rtc_endpoint_delete(_endpoint); _endpoint = nullptr;
    }
}

void Conn::init() {
    _start_time = Clock::now();
    // 0. init HttpClient
    if (!HttpClient::init(g_main_context_get_thread_default())) {
        dbge("init HttpClient failed");
        handler_.on_conn_init(this, false, handler_data_);
        return;
    }

    char rtc_id_str[64] = {0,};
    sprintf(rtc_id_str, "%d", rtc_id_);

    bool drop_audio = false;
    bool drop_video = false;

    // 1. allocate resources and initialize
    _endpoint = rtc_endpoint_create(g_main_context_get_thread_default(),
            rtc_id_str,
            ctx_.mediaAddress.c_str(),
            ctx_.minMediaPort, ctx_.maxMediaPort,
            false, // is dump rtp
            false, // is dump rtcp
            ctx_.video_bitrate / 1000,   // max video bitrate
            0,     // min video bitrate
            false, // is translating fec
            false, // is disable h264
            false, // is disable fec
            false,  // is compensate rtt
            false,  // is experiment esmb
            false,  // lost 0
            false,   // enable bwe
            true,     // force send sr
            false,  // send nack
            0,      // nack cache size
            drop_audio,
            drop_video
    );
    if (!_endpoint) {
        dbge("rtc_endpoint_create failed");
        handler_.on_conn_init(this, false, handler_data_);
        return;
    }

    rtc_endpoint_callbacks callbacks;
    callbacks.on_destroy = nullptr;
    callbacks.on_stream_data = &Conn::on_stream_data;
    callbacks.on_event = &Conn::on_stream_event;

    _callback_handle = rtc_endpoint_register_callback(_endpoint, this, &callbacks);

    _local_rtc_id = rtc_endpoint_get_id(_endpoint);

    auto now = Clock::now();
    next_stat_time_ = now;

    _last_video_key_frame_time = now;
    _video_seq_number = 1001;

    // 2. send initC signal to xswitch
    _remote_rtc_id.clear();
    _remote_sdp.clear();
    _remote_candidates.clear();

    // request formatter
    rapidjson::StringBuffer sb;
    rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
    writer.StartObject();
    writer.Key("op"); writer.String("initC");
    if (ctx_.mixer_id != nullptr) {
        writer.Key("ctxId"); writer.String(ctx_.mixer_id);
        writer.Key("confType"); writer.String("audio-mixer");
    }

    if (ctx_.sequence == Context::Sequence::Offerer) {
        std::list<std::string> hints;
        if (rtc_endpoint_generate_offer(_endpoint, hints)) {
            dbgw("init failed, rtc_endpoint_generate_offer error");
            handler_.on_conn_init(this, false, handler_data_);
            return;
        }
        int local_sdp_len = 0;
        const char* local_sdp = rtc_endpoint_get_local_sdp(_endpoint, &local_sdp_len);
        std::string sdp(local_sdp, local_sdp_len);
        writer.Key("sdp"); writer.StartObject();
        writer.Key("type"); writer.String("offer");
        writer.Key("sdp"); writer.String(sdp);
        writer.EndObject();
    }
    writer.EndObject();

    auto begin_time = Clock::now();
    async_post(get_rtc_url(), kContentTypeJson, std::string(sb.GetString(),
            sb.GetSize()), new HttpCallback([this, begin_time]
            (int status_code, const std::string& message) {
        dbgd("init completed, %d %s", status_code, message.c_str());
        if (status_code != 200) {
            dbgw("init failed, server replies %d", status_code);
            handler_.on_conn_init(this, false, handler_data_);
            return;
        }
        rapidjson::Document jsonDoc;
        jsonDoc.Parse(message.c_str());
        if (!jsonDoc.IsObject()) {
            dbgw("init failed, invalid json: %s", message.c_str());
            handler_.on_conn_init(this, false, handler_data_);
            return;
        }
        rapidjson::Value::ConstMemberIterator element = jsonDoc.FindMember("status");
        if (element == jsonDoc.MemberEnd() || !element->value.IsNumber()) {
            dbgw("init failed, invalid json: %s", message.c_str());
            handler_.on_conn_init(this, false, handler_data_);
            return;
        }
        int status = element->value.Get<int>();
        if (status) {
            dbgw("init failed, status: %d", status);
            handler_.on_conn_init(this, false, handler_data_);
            return;
        }

        element = jsonDoc.FindMember("sdp");
        if (element == jsonDoc.MemberEnd() || !element->value.IsObject()) {
            dbgw("init failed, invalid (sdp) json: %s", message.c_str());
            handler_.on_conn_init(this, false, handler_data_);
            return;
        }

        rapidjson::Value::ConstMemberIterator subElement = element->value.FindMember("sdp");
        if (subElement == element->value.MemberEnd() || !subElement->value.IsString()) {
            dbgw("init failed, invalid (sdp.sdp) json: %s", message.c_str());
            handler_.on_conn_init(this, false, handler_data_);
            return;
        }
        _remote_sdp = subElement->value.Get<std::string>();
        element = jsonDoc.FindMember("rtcId");
        if (element == jsonDoc.MemberEnd() || !element->value.IsString()) {
            dbgw("init failed, invalid json: %s", message.c_str());
            handler_.on_conn_init(this, false, handler_data_);
            return;
        }
        _remote_rtc_id = element->value.Get<std::string>();

        if (ctx_.sequence == Context::Sequence::Offerer) {
            if (rtc_endpoint_process_answer(this->_endpoint, _remote_sdp.c_str(), _remote_sdp.length(), 0)) {
                dbgw("init failed: processing remote answer");
                handler_.on_conn_init(this, false, handler_data_);
                return;
            }
        } else {
            if (rtc_endpoint_process_offer(this->_endpoint, _remote_sdp.c_str(), _remote_sdp.length())) {
                dbgw("init failed: processing remote offer");
                handler_.on_conn_init(this, false, handler_data_);
                return;
            }
        }

        element = jsonDoc.FindMember("candidates");
        if (element == jsonDoc.MemberEnd() || !element->value.IsArray()) {
            dbgw("init failed, invalid (candidates) json: %s", message.c_str());
            handler_.on_conn_init(this, false, handler_data_);
            return;
        }

        for (int k = 0; k < element->value.Size(); ++k) {
            if (!element->value[k].IsObject()) {
                dbgw("init failed, invalid (candidate) json: %s", message.c_str());
                handler_.on_conn_init(this, false, handler_data_);
                return;
            }
            auto& subElement = element->value[k];
            auto midElement = subElement.FindMember("mid");
            auto sdpElement = subElement.FindMember("candidate");
            auto mlineIndexElement = subElement.FindMember("mlineindex");

            if (midElement == subElement.MemberEnd() || sdpElement == subElement.MemberEnd() || mlineIndexElement == subElement.MemberEnd()) {
                dbgw("init failed, invalid (candidate member) json: %s", message.c_str());
                handler_.on_conn_init(this, false, handler_data_);
                return;
            }
            _remote_candidates.emplace_back(Candidate { std::string("a=") + sdpElement->value.Get<std::string>(), midElement->value.Get<std::string>(), mlineIndexElement->value.Get<int>()});
        }

        for (auto& c : _remote_candidates) {
            if (ctx_.replace_candidate) {
                auto from = c.sdp.find(ctx_.replace_candidate_from);
                if (from != std::string::npos) {
                    c.sdp.replace(from, ctx_.replace_candidate_from.length(), ctx_.replace_candidate_to);
                }
            }
            if (rtc_endpoint_add_remote_candidate(_endpoint, c.mlineIndex, c.mid.c_str(), c.sdp.c_str(), 0)) {
                dbgw("init failed, rtc_endpoint_add_remote_candidate failed");
                handler_.on_conn_init(this, false, handler_data_);
                return;
            }
        }

        handler_.on_conn_init(this, true, handler_data_);
        return;
    }));
}

void Conn::answer() {
    if (ctx_.sequence != Context::Sequence::Answerer) {
        dbge("answer called in non-answerer sequence");
        handler_.on_conn_answer(this, false, handler_data_);
        return;
    }
    if (rtc_endpoint_generate_answer(_endpoint, 0)) {
        dbgw("answer failed, rtc_endpoint_generate_answer error");
        handler_.on_conn_answer(this, false, handler_data_);
        return;
    }

    int local_sdp_len = 0;
    const char* local_sdp = rtc_endpoint_get_local_sdp(_endpoint, &local_sdp_len);
    std::string sdp(local_sdp, local_sdp_len);
    rapidjson::StringBuffer sb;
    rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
    writer.StartObject();
    writer.Key("op"); writer.String("ansC");
    writer.Key("rtcId"); writer.String(_remote_rtc_id);
    writer.Key("sdp"); writer.StartObject();
    writer.Key("type"); writer.String("answer");
    writer.Key("sdp"); writer.String(sdp);
    writer.EndObject();
    writer.EndObject();

    auto begin_time = Clock::now();
    async_post(get_rtc_url(), kContentTypeJson, std::string(sb.GetString(),
            sb.GetSize()), new HttpCallback([&,begin_time,this]
            (int status_code, const std::string& message) {
        auto cost_time = Clock::now() - begin_time;
//            _stats.update("time.answ", std::chrono::duration_cast<std::chrono::milliseconds>(cost_time).count());
        dbgd("answer completed %d %s", status_code, message.c_str());
        if (status_code != 200) {
            dbgw("answer failed, server replies %d", status_code);
            handler_.on_conn_answer(this, false, handler_data_);
            return;
        }

        rapidjson::Document jsonDoc;
        jsonDoc.Parse(message.c_str());
        if (!jsonDoc.IsObject()) {
            dbgw("answer failed, invalid json: %s", message.c_str());
            handler_.on_conn_answer(this, false, handler_data_);
            return;
        }
        auto element = jsonDoc.FindMember("status");
        if (element == jsonDoc.MemberEnd() || !element->value.IsNumber()) {
            dbgw("answer failed, no status element, json: %s", message.c_str());
            handler_.on_conn_answer(this, false, handler_data_);
            return;
        }
        int status = element->value.Get<int>();
        if (status) {
            dbgw("answer failed, status: %d", status);
            handler_.on_conn_answer(this, false, handler_data_);
            return;
        }

        handler_.on_conn_answer(this, true, handler_data_);
        return;
    }));
}

void Conn::trickle() {
    rapidjson::StringBuffer sb;
    rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
    writer.StartObject();
    writer.Key("op"); writer.String("tcklC");
    writer.Key("rtcId"); writer.String(_remote_rtc_id);
    writer.Key("candidates");
    writer.StartArray();
    for(int mlineindex = 0; mlineindex < rtc_endpoint_get_mline_count(_endpoint); mlineindex++){
        const char * mid = nullptr;
        int candidate_count = 0;
        char **candidates = nullptr;
        if(rtc_endpoint_get_candidates(_endpoint, 0, mlineindex, &mid, &candidate_count, &candidates)){
            dbgw("rtc_endpoint_get_candidates error");
            continue;
        }
        for(int ci = 0; ci < candidate_count; ci++){
            std::string candidate(&candidates[ci][2], strlen(candidates[ci]) - 2);
            writer.StartObject();
            writer.Key("candidate"); writer.String(candidate);
            writer.Key("sdpMid"); writer.String(mid);
            writer.Key("sdpMLineIndex"); writer.Int(mlineindex);
            writer.EndObject();
        }
    }
    writer.EndArray();
    writer.EndObject();

    auto begin_time = Clock::now();
    async_post(get_rtc_url(), kContentTypeJson, std::string(sb.GetString(),
            sb.GetSize()), new HttpCallback([&,this,begin_time]
            (int status_code, const std::string& message) {
        if (status_code != 200) {
            dbgw("tcklC failed, server replies %d", status_code);
            handler_.on_conn_trickle(this, false, handler_data_);
            return;
        }

        rapidjson::Document jsonDoc;
        jsonDoc.Parse(message.c_str());
        if (!jsonDoc.IsObject()) {
            dbgw("tcklC failed, invalid json: %s", message.c_str());
            handler_.on_conn_trickle(this, false, handler_data_);
            return;
        }
        auto element = jsonDoc.FindMember("status");
        if (element == jsonDoc.MemberEnd() || !element->value.IsNumber()) {
            dbgw("tcklC failed invalid json: %s", message.c_str());
            handler_.on_conn_trickle(this, false, handler_data_);
            return;
        }
        int status = element->value.Get<int>();
        if (status) {
            dbgw("tcklC failed, status: %d", status);
            handler_.on_conn_trickle(this, false, handler_data_);
            return;
        }

        handler_.on_conn_trickle(this, true, handler_data_);
        return;
    }));
}

void Conn::terminate() {
    _terminating = true;
    rapidjson::StringBuffer sb;
    rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
    writer.StartObject();
    writer.Key("op"); writer.String("termC");
    writer.Key("rtcId"); writer.String(_remote_rtc_id);
    writer.EndObject();

    auto begin_time = Clock::now();
    async_post(get_rtc_url(), kContentTypeJson, std::string(sb.GetString(),
            sb.GetSize()), new HttpCallback([&,begin_time,this]
            (int status_code, const std::string& message) {
        _terminated = true;
        dbgd("termC completed %d %s", status_code, message.c_str());
        if (status_code != 200) {
            dbgw("termC failed, server replies %d", status_code);
            handler_.on_conn_terminate(this, false, handler_data_);
            return;
        }

        rapidjson::Document jsonDoc;
        jsonDoc.Parse(message.c_str());
        if (!jsonDoc.IsObject()) {
            dbgw("termC failed, invalid json: %s", message.c_str());
            handler_.on_conn_terminate(this, false, handler_data_);
            return;
        }
        auto element = jsonDoc.FindMember("status");
        if (element == jsonDoc.MemberEnd() || !element->value.IsNumber()) {
            dbgw("termC failed, no status element, json: %s", message.c_str());
            handler_.on_conn_terminate(this, false, handler_data_);
            return;
        }
        int status = element->value.Get<int>();
        if (status) {
            dbgw("termC failed, status: %d", status);
            handler_.on_conn_terminate(this, false, handler_data_);
            return;
        }

        handler_.on_conn_terminate(this, true, handler_data_);
        return;
    }));
}

std::string Conn::get_rtc_url() const {
    // if we reach here _ctx.rtcAddresses.size() must be 1
    return std::string("http://" + ctx_.rtcAddresses[0] + "/rtc");
}

void Conn::subscribe() {
    rapidjson::StringBuffer sb;
    rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
    writer.StartObject();
    writer.Key("op"); writer.String("subC");
    writer.Key("rtcId"); writer.String(_remote_rtc_id);

    if (pub_rtc_id_ != nullptr) {
        g_message("rtc %s subscribe %s", _remote_rtc_id.c_str(),
                pub_rtc_id_);
        writer.Key("pubRtcId"); writer.String(pub_rtc_id_);
    } else {
        g_message("rtc %s subscribe self", _remote_rtc_id.c_str());
        writer.Key("pubRtcId"); writer.String(_remote_rtc_id);
    }

    if (ctx_.mixer_id != nullptr) {
        writer.Key("audio");
        writer.Bool(false);
        writer.Key("video");
        writer.Bool(true);
    }

    writer.EndObject();
    auto begin_time = Clock::now();
    async_post(get_rtc_url(), kContentTypeJson, std::string(sb.GetString(),
            sb.GetSize()), new HttpCallback([&,this,begin_time]
            (int status_code, const std::string& message) {
        if (status_code != 200) {
            dbgw("subC failed, server replies %d", status_code);
            handler_.on_conn_subscribe(this, false, handler_data_);
            return;
        }

        rapidjson::Document jsonDoc;
        jsonDoc.Parse(message.c_str());
        if (!jsonDoc.IsObject()) {
            dbgw("subC failed, invalid json: %s", message.c_str());
            handler_.on_conn_subscribe(this, false, handler_data_);
            return;
        }
        auto element = jsonDoc.FindMember("status");
        if (element == jsonDoc.MemberEnd() || !element->value.IsNumber()) {
            dbgw("subC failed invalid json: %s", message.c_str());
            handler_.on_conn_subscribe(this, false, handler_data_);
            return;
        }
        int status = element->value.Get<int>();
        if (status) {
            dbgw("subC failed, status: %d", status);
            handler_.on_conn_subscribe(this, false, handler_data_);
            return;
        }

        handler_.on_conn_subscribe(this, true, handler_data_);
        return;
    }));
}

void Conn::send_packets() {
    if (!_setup) {
        return;
    }
    auto now = Clock::now();
    auto now_epoch_ms = epochMs(now);

    if (_resched_video) {
        _next_video_frame_time = now;
        _resched_video = false;
    }
    if (_resched_audio) {
        next_audio_frame_time_ = now;
        _resched_audio = false;
    }
    while (now >= _next_video_frame_time && _video_frame_size > 0) {
        bool is_key_frame = false;
        if (now - _last_video_key_frame_time > kVideoKeyframePeriod) {
            is_key_frame = true;
            _last_video_key_frame_time = now;
        }

        int frame_size = _video_frame_size;
        while (frame_size > kMaxPacketSize) {
            send_video_packet(is_key_frame, false, kMaxPacketSize, now_epoch_ms);
            frame_size -= kMaxPacketSize;
        }
        send_video_packet(is_key_frame, true, frame_size, now_epoch_ms);

        _next_video_frame_time += kVideoFramePeriod;
    }

    if (audio_) {
        while (now >= next_audio_frame_time_) {
            sendAudioPacket(now_epoch_ms);
            next_audio_frame_time_ += kAudioFramePeriod;
        }
    } else {
        while (now >= next_audio_frame_time_) {
            sendDummyAudioPacket(now_epoch_ms);
            next_audio_frame_time_ += kAudioFramePeriod;
        }
    }
}

void Conn::get_send_recv_bytes(uint64_t &send, uint64_t &recv) {
    rtc_endpoint_reset_bytes(_endpoint, &send, &recv);
}

void Conn::get_send_recv_packets(uint64_t &send, uint64_t &recv) {
    rtc_endpoint_reset_packets(_endpoint, &send, &recv);
}

void Conn::send_video_packet(bool is_key_frame, bool is_marker, int size,
        uint64_t ms) {
    // fake a 12byte rtp header
    uint16_t seq = _video_seq_number++;
    uint32_t ts = (kVideoSampleRate / 1000) * ms;
    buf_[0] = 0x80;
    buf_[1] = (is_marker ? 0x80 : 0) | kPayloadTypeVp8;

    be_set_u16(seq, buf_ + 2);
    be_set_u32(ts, buf_ + 4);
    be_set_u32(kVideoSSRC, buf_ + 8);

    unsigned char* rtp_payload = buf_ + 12;
    unsigned char* vp8_payload_descriptor = rtp_payload;
    *vp8_payload_descriptor = 0x10; // S = 1, others 0
    unsigned char* vp8_payload_header = vp8_payload_descriptor + 1;
    *vp8_payload_header = is_key_frame ? 0x00 : 0x01;
    rtc_endpoint_send_stream_data(_endpoint, RTC_VIDEO, rtc_codec_id_vp8,
            rtc_codec_id_unknown, // fec_codec
            0,  // is_rtcp
            0,  // is_rtx
            buf_, 12 + size);

}

void Conn::sendDummyAudioPacket(uint64_t now) {
    uint16_t seq = audio_seq_number_++;
    uint32_t ts = audio_rtp_timestamp_;
    audio_rtp_timestamp_ += std::chrono::duration_cast<std::chrono::milliseconds>(kAudioFramePeriod).count()* kAudioSampleRate / 1000;

    buf_[0] = 0x90; // V=2 X=1 P=0 CC=0
    buf_[1] = 0x80 | kPayloadTypeOpus;
    be_set_u16(seq, buf_ + 2);
    be_set_u32(ts, buf_ + 4);
    be_set_u32(kAudioSSRC, buf_ + 8);

    size_t packet_buffer_len = 12;

    buf_[packet_buffer_len+0] = 0xBE;
    buf_[packet_buffer_len+1] = 0xDE;
    buf_[packet_buffer_len+2] = 0x00;
    buf_[packet_buffer_len+3] = 0x01;
    buf_[packet_buffer_len+4] = RtpExtensionType ::SSRCAudioLevel << 4 | 0;
    buf_[packet_buffer_len+5] = 127;
    buf_[packet_buffer_len+6] = 0x00;
    buf_[packet_buffer_len+7] = 0x00;
    packet_buffer_len += 8;
    packet_buffer_len += 40;
    rtc_endpoint_send_stream_data(_endpoint,
            RTC_AUDIO,
            rtc_codec_id_opus,
            rtc_codec_id_unknown, // fec_codec
            0,  // is_rtcp
            0,  // is_rtx
            (unsigned char *)buf_,
            packet_buffer_len);
}

void Conn::sendAudioPacket(uint64_t now) {
    uint16_t seq = audio_seq_number_++;
    uint32_t ts = audio_rtp_timestamp_;
    audio_rtp_timestamp_ += std::chrono::duration_cast<std::chrono::milliseconds>(kAudioFramePeriod).count() * kAudioSampleRate / 1000;

    buf_[0] = 0x90; // V=2 X=1 P=0 CC=0
    buf_[1] = 0x80 | kPayloadTypeOpus;
    be_set_u16(seq, buf_ + 2);
    be_set_u32(ts, buf_ + 4);
    be_set_u32(kAudioSSRC, buf_ + 8);

    audio_packet_id_++;
    if (audio_packet_id_ >= audio_->Packets()) {
        audio_packet_id_ = 0;
    }

    auto packet = audio_->Packet(audio_packet_id_);
    if (!packet) {
        return;
    }
    size_t packet_buffer_len = 12;

    buf_[packet_buffer_len+0] = 0xBE;
    buf_[packet_buffer_len+1] = 0xDE;
    buf_[packet_buffer_len+2] = 0x00;
    buf_[packet_buffer_len+3] = 0x01;
    buf_[packet_buffer_len+4] = RtpExtensionType ::SSRCAudioLevel << 4 | 0;
    buf_[packet_buffer_len+5] = (-audio_->Level(audio_packet_id_)) & 0x7F;
    buf_[packet_buffer_len+6] = 0x00;
    buf_[packet_buffer_len+7] = 0x00;
    packet_buffer_len += 8;

    memcpy(buf_ + packet_buffer_len, packet->data, packet->size);
    packet_buffer_len += packet->size;
    rtc_endpoint_send_stream_data(_endpoint,
            RTC_AUDIO,
            rtc_codec_id_opus,
            rtc_codec_id_unknown, // fec_codec
            0,  // is_rtcp
            0,  // is_rtx
            (unsigned char *)buf_,
            packet_buffer_len);
}

void Conn::handle_setup_event() {
    if (!_setup) {
        _setup = true;

        auto setup_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
                Clock::now() - _start_time).count();

        metric_.setup_time.Update(setup_ms);
        handler_.on_conn_setup(this, setup_ms, handler_data_);
    }
}

void Conn::handle_incoming_rtcp(uint8_t *buf, int size) {
    // parse rtcp to get report
    int pkttype = rtcp_parse_first(rtcp_parser_, buf, size);
    const RTCPPacket * body = rtcp_get_body(rtcp_parser_);
    int report_nb = 0;
    while( pkttype  != kInvalid){
        switch(pkttype){
            case kReportBlockItem: {
                uint32_t media_ssrc = body->ReportBlockItem.SSRC;
                uint32_t hseq = body->ReportBlockItem.ExtendedHighestSequenceNumber;
                uint32_t lsr = body->ReportBlockItem.LastSR;
                uint32_t dslr = body->ReportBlockItem.DelayLastSR;
                uint8_t lost_frac = body->ReportBlockItem.FractionLost;
                uint32_t lost_cum = body->ReportBlockItem.CumulativeNumOfPacketsLost;
                uint32_t jitter = body->ReportBlockItem.Jitter;
                int64_t rcv_ts = get_timestamp_ms();
                uint32_t now = ms_to_ntp_mid32(rcv_ts);
                uint32_t rtt_ntp_mid32 = 0;

                if (lsr != 0) {
                    if (now >= lsr) {
                        rtt_ntp_mid32 = now - lsr;
                    } else {
                        // handling wrapped now as follows:
                        rtt_ntp_mid32 = 0xFFFFFFFFu - lsr + now + 1;
                    }
                    if (rtt_ntp_mid32 > dslr) {
                        rtt_ntp_mid32 -= dslr;
                    } else {
                        rtt_ntp_mid32 = 0;
                    }

                    int rtt_ms = (rtt_ntp_mid32 * 1000) >> 16;
                    if(rtt_ms <= 10000){
                        metric_.rtt.Update(rtt_ms);
                    }
                }

                if (media_ssrc == kAudioSSRC) {
                    metric_.jitter.Update(jitter / (kAudioSampleRate / 1000));
                } else if (media_ssrc == kVideoSSRC) {
                    metric_.jitter.Update(jitter / (kVideoSampleRate / 1000));
                }

//  因为xswitch不计算丢包，所以算出的丢包永远是0
//                    if (media_ssrc == kAudioSSRC) {
//                        if (rtcp_audio_last_hseq_) {
//                            uint32_t d = hseq - rtcp_audio_last_hseq_;
//                            uint32_t loss = lost_frac * d / 255;
//                            //printf("audio loss %d\n", loss);
//                            _ctx.s_loss.Update(lost_frac * d / 255);
//                        }
//                        rtcp_audio_last_hseq_ = hseq;
//                        //printf("audio hseq %d loss_frac %d\n", hseq, lost_frac);
//                    } else if (media_ssrc == kVideoSSRC) {
//
//                        if (rtcp_video_last_hseq_) {
//                            uint32_t d = hseq - rtcp_video_last_hseq_;
//                            uint32_t loss = lost_frac * d / 255;
//                            //printf("video loss %d\n", loss);
//                            _ctx.s_loss.Update(lost_frac * d / 255);
//                        }
//                        rtcp_video_last_hseq_ = hseq;
//                        //printf("video hseq %d loss_frac %d\n", hseq, lost_frac);
//                    }
                report_nb -= 1;
                if (report_nb > 0) {
                    pkttype = rtcp_parse_next_item(rtcp_parser_);
                } else {
                    pkttype = rtcp_parse_next_packet(rtcp_parser_);
                }
            }
                break;
            case kSr:
                if (body->SR.NumberOfReportBlocks > 0) {
                    report_nb = body->SR.NumberOfReportBlocks;
                    pkttype = rtcp_parse_next_item(rtcp_parser_);
                } else {
                    pkttype = rtcp_parse_next_packet(rtcp_parser_);
                }
                break;
            case kRr:
                if (body->RR.NumberOfReportBlocks > 0) {
                    report_nb = body->RR.NumberOfReportBlocks;
                    pkttype = rtcp_parse_next_item(rtcp_parser_);
                } else {
                    pkttype = rtcp_parse_next_packet(rtcp_parser_);
                }
                break;
            default:
                pkttype = rtcp_parse_next_packet(rtcp_parser_);
                break;
        }
    }
}

void Conn::handle_incoming_rtp(uint8_t *buf, int size) {
}
